package provisioner;

import provisioner.util.OSShell;

import java.net.Inet4Address;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.InterfaceAddress;
import java.net.NetworkInterface;
import java.security.Security;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.regex.MatchResult;

import com.xerox.amazonws.ec2.EC2Exception;
import com.xerox.amazonws.ec2.Jec2;
import com.xerox.amazonws.ec2.InstanceStateChangeDescription;
import com.xerox.amazonws.ec2.InstanceType;
import com.xerox.amazonws.ec2.LaunchConfiguration;
import com.xerox.amazonws.ec2.ReservationDescription;
import com.xerox.amazonws.ec2.ReservationDescription.Instance;

import com.jcraft.jsch.*;

public class EucaProvider implements ComputeProvider, ProvisionEventNotifier, Runnable {

   private String vpnServerIp = "203.185.96.116";
   private String privateIp = "192.168.1.1";
   private String privateNetwork = "192.168.1.0";
   private String privateNetmask = "255.255.255.0";
   private int STATUS_PERIOD = 15*1000; // milliseconds.
   private int MAX_INSTANCE = 6;
   private Map<String,EucaVmNode> computeMap = new HashMap<String,EucaVmNode>();
   private Set<EucaVmNode> instancesWaitingInit = new HashSet<EucaVmNode>();
   private Map<EucaVmNode,Set<ProvisionEventListener>> listenerSetMap;

   private String keyName;
   private String privateKeyFile;
   private String amiId;
   private InstanceType instanceType;
   private String kernelId;
   private String ramdiskId;

   private String CAKeyFile;
   private String CACertFile;

   private Jec2 ec2;

   public EucaProvider(String amiId, String privateKeyFile, String CAKeyFile, String CACertFile) throws Exception {
      this(amiId,privateKeyFile,CAKeyFile,CACertFile,"m1.small","","");
   }

   public EucaProvider(String amiId, String privateKeyFile, String CAKeyFile, String CACertFile, String instanceType) throws Exception {
      this(amiId,privateKeyFile,CAKeyFile,CACertFile,instanceType,"","");
   }

   public EucaProvider(String amiId, String privateKeyFile, String CAKeyFile, String CACertFile, String instanceType, String kernelId, String ramdiskId) throws Exception {
      this.amiId = amiId;
      this.privateKeyFile = privateKeyFile;
      this.CAKeyFile = CAKeyFile;
      this.CACertFile = CACertFile;
      this.instanceType = InstanceType.getTypeFromString(instanceType);
      this.kernelId = kernelId;
      this.ramdiskId = ramdiskId;

      Properties props = new Properties();
//      props.load(EucaProvider.class.getClassLoader().getResourceAsStream("aws.properties"));
//      ec2 = new Jec2(props.getProperty("aws.accessId"), props.getProperty("aws.secretKey"),false,"192.0.3.23",8773);
//      this.keyName = props.getProperty("aws.keyName");
      ec2 = new Jec2("iqExU7MEpStSGUk6mpqIyHbMHIyabUeJ3PPSTg","ZjfU3mEGVhB0yuuCbliCfBw8TWeHsJSJ1vtnQ",false,"192.0.3.23",8773);
      System.out.println(EucaProvider.class.getName() + " limits " + MAX_INSTANCE + " instances");
      this.keyName = "mykey";
      ec2.setResourcePrefix("/services/Eucalyptus");
      ec2.setSignatureVersion(1);

      this.listenerSetMap = new HashMap<EucaVmNode,Set<ProvisionEventListener>>();

      init();
   }

   /*
    * Assume private network in on eth0, according to Rocks cluster
    */
   private void init() throws Exception {
      NetworkInterface eth0 = NetworkInterface.getByName("eth0");
      if (eth0 == null) {
         throw new Exception("initialization failed: no network configuration found");
      }
      List<InterfaceAddress> ifaddrs = eth0.getInterfaceAddresses();
      InterfaceAddress ifaddr = null;
      for (InterfaceAddress i : ifaddrs) {
         InetAddress iaddr = i.getAddress();
         if (iaddr instanceof Inet4Address) {
            ifaddr =  i;
            break;
         }
      }
      if (ifaddr == null) {
         throw new Exception("initialization failed: no network configuration found");
      }
      InetAddress ip = ifaddr.getAddress();
      InetAddress broadcast = ifaddr.getBroadcast();
      int prefix = ifaddr.getNetworkPrefixLength();
      int bits = 0xffffffff << (32 - prefix);
      byte[] bytes = new byte[]{ (byte)(bits >>> 24), (byte)(bits >> 16 & 0xff), (byte)(bits >> 8 & 0xff), (byte)(bits & 0xff) };
      InetAddress mask = InetAddress.getByAddress(bytes);
      byte[] ipBytes = ip.getAddress();
      byte[] maskBytes = mask.getAddress();
      byte[] netBytes = new byte[ipBytes.length];
      for (int i = 0; i < ipBytes.length; i++) {
         netBytes[i] = (byte) (ipBytes[i] & maskBytes[i]);
      }
      InetAddress net = InetAddress.getByAddress(netBytes);
      privateIp = ip.getHostAddress();
      privateNetmask = mask.getHostAddress();
      privateNetwork = net.getHostAddress();

      Thread t = new Thread(this);
      t.start();

   }

   public synchronized ComputeNode startNode(String properties) throws Exception {
      return startNode(properties,this.instanceType);
   }

   public synchronized ComputeNode startNode(String properties, String instanceType) throws Exception {
      return startNode(properties,InstanceType.getTypeFromString(instanceType));
   }

   public synchronized ComputeNode startNode(String properties, InstanceType instanceType) throws Exception {
      EucaVmNode node = null;
      if (computeMap.size() >= MAX_INSTANCE) {
         throw new NoMoreResourceException("Maximum number of instances reached");
      }
      Instance instance = null;
      try {
         instance = runInstance(properties, instanceType); // now assume properties == image id
      } catch (EC2Exception e) {
         throw new ResourceNotMatchedException(e);
      }
      if (instance != null) {
         node = new EucaVmNode(instance.getInstanceId(),ComputeNode.STARTING); 
         node.setProperty(properties);  // only single property is supported now.
         node.setContext(vpnServerIp,privateIp,privateNetwork,privateNetmask, privateKeyFile,CACertFile);
         computeMap.put(node.getId(),node);
System.out.println("# of instances now is " + computeMap.size());
      }
      return node;
   }

   public synchronized void stopNode(String nodeId) throws Exception {
         EucaVmNode node = computeMap.get(nodeId); 
         if (node == null) {
            System.err.println("Cannot stop node " + nodeId + ": node not found or in our control");
            return;
         }
         System.out.println("Stopping node " + nodeId);
         terminateInstance(nodeId);
         instancesWaitingInit.remove(node); 
         if (node.getState() == ComputeNode.READY) {
            try {
               removeRocksHost(node);
            } catch (Exception e) {
               System.err.println("Error occurs during removing node " + nodeId); 
            }
         }
         computeMap.remove(nodeId);
 System.out.println("# of instances now is " + computeMap.size());
         Set<ProvisionEventListener> listeners = listenerSetMap.get(node);
         if (listeners != null) {
            notifyDownStatus(node,listeners);
            listenerSetMap.remove(node);
         }
   }

   public synchronized void stopNode(ComputeNode node) throws Exception {
      if (node != null) {
         stopNode(node.getId());
      }
   }

   public synchronized Collection<ComputeNode> getComputeNodes() throws Exception {
      return new ArrayList<ComputeNode>(computeMap.values());
   }

   public void run() {
      while (true) {
         try {
            updateStatus();
         } catch (Exception e) {
            e.printStackTrace();
         }
         try {
            Thread.sleep(STATUS_PERIOD);
         } catch (InterruptedException e) {

         }

      }
   }

   public synchronized void registerListener(Resource resource, ProvisionEventListener listener) throws Exception {
      String id = resource.getId(); 
      EucaVmNode node = computeMap.get(id); 
      if (node == null) {
         System.err.println("EucaVmNode " + id + " not found for event listener");
         return;
      }
      Set<ProvisionEventListener> listeners = listenerSetMap.get(node);
      if (listeners == null) {
         listeners = new HashSet<ProvisionEventListener>();
         listenerSetMap.put(node,listeners);
      }
      listeners.add(listener);
   }

   public synchronized void unregisterListener(Resource resource, ProvisionEventListener listener) throws Exception {
      String id = resource.getId(); 
      EucaVmNode node = computeMap.get(id); 
      if (node == null) {
         System.err.println("EucaVmNode " + id + " not found for event listener");
         return;
      }
      Set<ProvisionEventListener> listeners = listenerSetMap.get(node);
      if (listeners == null) {
         return;
      }
      listeners.remove(listener);
   }

   private synchronized void notifyUpStatus(EucaVmNode node, Set<ProvisionEventListener> listeners) throws Exception {
      Collection<Resource> resources = new ArrayList<Resource>();
      resources.add(node);
      ResourceStatusUpEvent e = new ResourceStatusUpEvent(resources);
      for (ProvisionEventListener listener : listeners) {
         listener.actionPerformed(e);
      }
   }

   private synchronized void notifyDownStatus(EucaVmNode node, Set<ProvisionEventListener> listeners) throws Exception {
      Collection<Resource> resources = new ArrayList<Resource>();
      resources.add(node);
      ResourceStatusDownEvent e = new ResourceStatusDownEvent(resources);
      for (ProvisionEventListener listener : listeners) {
         listener.actionPerformed(e);
      }
   }

   public synchronized Instance findInstance(String instanceId) throws Exception {
      String[] instanceIds = { instanceId };
      List<ReservationDescription> descriptions = ec2.describeInstances(instanceIds);
      Instance instance = null;
      for (Instance inst : descriptions.get(0).getInstances()) {
         //System.out.println(instance.getInstanceId() + "," + instance.getDnsName() + "," + instance.getState());
         if (inst.getInstanceId().equals(instanceId)) {
            instance = inst; 
            break;
         }
      }
      return instance;
   }

   private synchronized Instance runInstance(String amiId) throws Exception {
      return runInstance(amiId,this.instanceType);
   }

   private synchronized Instance runInstance(String amiId, String instanceType) throws Exception {
      return runInstance(amiId,InstanceType.getTypeFromString(instanceType));
   }

   private synchronized Instance runInstance(String amiId, InstanceType instanceType) throws Exception {

      LaunchConfiguration conf = null;
      if ( (amiId == null) || (amiId.equals("")) ) {
         conf = new LaunchConfiguration(this.amiId);
      } else {
         conf = new LaunchConfiguration(amiId);
      }
      conf.setInstanceType(instanceType);
      conf.setKeyName(this.keyName);
System.out.println("Launching " + amiId + " " + this.instanceType + " with key " + this.keyName);
      ReservationDescription reservation=ec2.runInstances(conf);
      List<Instance> e_instances=reservation.getInstances();
      if(e_instances!=null && e_instances.size()>0) {
         Instance instance = e_instances.get(0); 
         return instance;
      }
      return null;
   }

   public synchronized void updateStatus() throws Exception {
      //System.out.println("updating instances' status ...");
      List<ReservationDescription> descriptions = ec2.describeInstances(new ArrayList<String>());
      Set<String> eucaInstances = new HashSet<String>();

      for (ReservationDescription description : descriptions) {
         for (Instance instance : description.getInstances()) {
            String instanceId = instance.getInstanceId();
            eucaInstances.add(instanceId);
            EucaVmNode node = computeMap.get(instanceId);
            if (node != null) { // take care only instances created by us
               String state = instance.getState();
               if (state.equals("running") && node.getState() == ComputeNode.STARTING) {
                  node.setHostname(instance.getDnsName()); // EUCA returns IP!
                  node.setIpAddress(instance.getDnsName());
                  node.setEucaIpAddress(instance.getPrivateIpAddress());
                  node.setState(ComputeNode.STARTED);
                  initializeInstance(node);
                  instancesWaitingInit.add(node);
               } else if (state.equals("running") && node.getState() == ComputeNode.CONTEXUALIZED) {
                  addRocksHost(node); // update rocks database
                  instancesWaitingInit.remove(node); 
                  System.out.println(node.getId() + " initialized in " + (System.currentTimeMillis()-node.getStartTime())/1000 + " s.");

                  Set<ProvisionEventListener> listeners = listenerSetMap.get(node);
                  if (listeners != null) {
                     notifyUpStatus(node,listeners);
                  }
                  node.setState(ComputeNode.READY);
                  node.getWorkingThread().join();
               }
            } else {
               System.err.println("Found instance " + instanceId + " not owned by us");
            }
         }
      }

      /*
       * remove dead nodes
       */
      Set<String> deadNodes = new HashSet<String>(computeMap.keySet());
      deadNodes.removeAll(eucaInstances);
      for (String deadNode : deadNodes) {
         System.err.println("Instance " + deadNode + " is found dead!!!");
         stopNode(deadNode);        
      }
   
   }

   private synchronized boolean terminateInstance(String instanceId) throws Exception {
      List<String> instanceIds = new ArrayList<String>();
      instanceIds.add(instanceId);
		
      List<InstanceStateChangeDescription> descriptions=ec2.terminateInstances(instanceIds);
      for(InstanceStateChangeDescription des : descriptions){
         if(des.getInstanceId().equals(instanceId)) {
            
            return true;
         }
      }
      return false;		
   }

   private void initializeInstance(EucaVmNode node) throws Exception {
      String id = node.getId();
      System.out.println("Initializing node " + id + "(see /tmp/" + id + ".log for details)");
      Thread nodeThread = new Thread(node);
      node.setWorkingThread(nodeThread);
      nodeThread.start(); // execute a thread
   } 

   private void addRocksHost(EucaVmNode node) throws Exception {
      String cmd = "/opt/rocks/sbin/insert-ethers --mac None --ipaddr " + node.getVpnIpAddress() + " --hostname " + node.getId() + " --batch --appliance Compute";
System.out.println(cmd);
      OSShell.exec(cmd);
      OSShell.exec("/opt/rocks/sbin/insert-ethers --update");
      
      // qmgr must use command array style
      String property = node.getProperty();
      String[] cmds = {"qmgr","-c","create node " + node.getId() + " properties+=" + property + ",properties+=provisioned"};
      OSShell.exec(cmds);
   }

   private void removeRocksHost(EucaVmNode node) throws Exception {
      OSShell.exec("/opt/rocks/sbin/insert-ethers --remove " + node.getId());
      OSShell.exec("/opt/rocks/sbin/insert-ethers --update");
      String[] cmds = {"qmgr","-c","delete node " + node.getId()};
      OSShell.exec(cmds);
      System.err.println("Rocks host " + node.getId() + " removed");
   }

   public static void main(String[] args) throws Exception {
      Security.addProvider(new org.bouncycastle.jce.provider.BouncyCastleProvider());
      EucaProvider provisioner = new EucaProvider("emi-CC711091","/home/ekasit/src/typica-1.7/mykey.private","/home/ekasit/src/typica-1.7/ca.key","/home/ekasit/src/typica-1.7/ca.crt");
      ComputeNode node = provisioner.startNode("emi-25E6125D","c1.medium");
      System.out.println("Starting instance " + node.getId());

      //provisioner.stopNode(node); 
        // testSingleNode(provider);
        //System.exit(0);
   }


   public static void testSingleNode(EucaProvider provisioner) throws Exception {
System.out.println("begin");
long t1 = System.currentTimeMillis();
        EucaVmNode node = new EucaVmNode("i-49250783");
        node.setIpAddress("192.0.3.202");
        JSch jsch = new JSch();
        jsch.addIdentity(provisioner.privateKeyFile);
        Session sshSession = jsch.getSession("root",node.getHostname());
        sshSession.setConfig("StrictHostKeyChecking","no");
        sshSession.connect(EucaVmNode.SSH_TIMEOUT);
        node.setSshSession(sshSession);
long t2 = System.currentTimeMillis();
System.out.println("create ssh " + (t2-t1)/1000 + " s.");
//t1 = t2;
//        node.createVpn(); 
//t2 = System.currentTimeMillis();
//System.out.println("createVpn " + (t2-t1)/1000 + " s.");
t1 = t2;
        node.sync411Config();
        node.setVpnIpAddress(node.checkVpnClientIp());
t2 = System.currentTimeMillis();
System.out.println("sync 411 " + (t2-t1)/1000 + " s.");
//t1 = t2;
//        provisioner.addRocksHost(node);
//t2 = System.currentTimeMillis();
//System.out.println("add rocks host " + (t2-t1)/1000 + " s.");
t1 = t2;
        node.runInitScript();
t2 = System.currentTimeMillis();
System.out.println("run initScript " + (t2-t1)/1000 + " s.");
System.out.println("done");
      
      //provisioner.removeRocksHost(node);

   }
}
